package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;

/**
 * 2. 实时统计每10分钟各个页面浏览量
 * 	条件：
 * 		page 页面日志
 * 		page_id 页面ID不能为空
 * 	窗口：
 * 		每10分钟
 * 	分组：
 * 		各个页面
 * 		 {
 * 		  "common": {
 * 		    "ar": "440000",
 * 		    "ba": "iPhone",
 * 		    "ch": "Appstore",
 * 		    "is_new": "1",
 * 		    "md": "iPhone Xs",
 * 		    "mid": "mid_409637",
 * 		    "os": "iOS 13.3.1",
 * 		    "uid": "364",
 * 		    "vc": "v2.1.134"
 *          },
 * 		  "page": {
 * 		    "during_time": 14631,
 * 		    "last_page_id": "home",
 * 		    "page_id": "mine"
 *          },
 * 		  "ts": 1713404821000
 *        }
 * 	条件：
 * 		page 页面日志
 * 		page_id 页面ID不能为空
 * @author xuanyu
 * @date 2025/10/21
 */
public class _02RealtimeWindowPageViewDemo {

	public static void main(String[] args) throws Exception{
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2. 数据源-source
		KafkaSource<String> source = KafkaSource.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setTopics("topic-log")
			.setGroupId("my-group-1")
			.setStartingOffsets(OffsetsInitializer.earliest())
			.setValueOnlyDeserializer(new SimpleStringSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			source, WatermarkStrategy.noWatermarks(), "Kafka Source"
		);
		//stream.print("kafka");

		// 3. 数据转换-transformation

		// todo 3-1. 指定数据事件时间和水位线Watermark
		SingleOutputStreamOperator<String> stream31 = stream.assignTimestampsAndWatermarks(
			WatermarkStrategy
				.<String>forBoundedOutOfOrderness(Duration.ofSeconds(5))
				.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
					@Override
					public long extractTimestamp(String element, long recordTimestamp) {
						// 解析json
						JSONObject jsonObject = JSON.parseObject(element);
						// 获取时间戳
						Long tsValue = jsonObject.getLong("ts");
						// 返回
						return tsValue;
					}
				})
		);


		// 3-2. 获取page页面日志，并且page_id 不为null 数据
		SingleOutputStreamOperator<String> stream32 = stream31.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					// 解析json
					JSONObject jsonObject = JSON.parseObject(value);
					// 获取page字段值
					Object page = jsonObject.get("page");
					if(null != page){
						// 获取page_id字段值
						JSONObject pageJsonObject = jsonObject.getJSONObject("page");
						String pageIdValue = pageJsonObject.getString("page_id");
						if(null != pageIdValue){
							return true ;
						}
					}
					// 不满足条件，返回false
					return false;
				}
			}
		);
		//stream31.print("page");

		// 3-3. 提取字段值
		SingleOutputStreamOperator<PageViewBean> stream33 = stream32.map(
			new MapFunction<String, PageViewBean>() {
				@Override
				public PageViewBean map(String value) throws Exception {
					// 解析json
					JSONObject jsonObject = JSON.parseObject(value);
					JSONObject pageJsonObject = jsonObject.getJSONObject("page");
					String pageIdValue = pageJsonObject.getString("page_id");
					// 返回
					return new PageViewBean(pageIdValue, 1L);
				}
			}
		);

		// 3-4. 按照页面分组
		KeyedStream<PageViewBean, String> stream34 = stream33.keyBy(
			new KeySelector<PageViewBean, String>() {
				@Override
				public String getKey(PageViewBean value) throws Exception {
					return value.getPageId();
				}
			}
		);

		// 3-5. 设置窗口：10分钟
		WindowedStream<PageViewBean, String, TimeWindow> stream35 = stream34.window(
			TumblingEventTimeWindows.of(Time.minutes(10))
		);

		// 3-6. 窗口计算
		SingleOutputStreamOperator<String> stream36 = stream35.apply(
			new WindowFunction<PageViewBean, String, String, TimeWindow>() {
				private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
				@Override
				public void apply(String pageId,
				                  TimeWindow window,
				                  Iterable<PageViewBean> input,
				                  Collector<String> out) throws Exception {
					// 窗口开始时间和结束时间
					String start = format.format(window.getStart());
					String end = format.format(window.getEnd());
					// 求和
					long pv = 0L;
					for (PageViewBean pageViewBean : input) {
						pv += pageViewBean.getPageView() ;
					}
					// 输出
					String output = start + "," + end + "," + pageId +"," + pv ;
					out.collect(output);
				}
			}
		);
		//stream36.print();
		/*
			2024-04-18 10:00:00,2024-04-18 10:10:00,mine,126
			2024-04-18 10:00:00,2024-04-18 10:10:00,register,135
			2024-04-18 10:00:00,2024-04-18 10:10:00,good_list,458
			2024-04-18 10:00:00,2024-04-18 10:10:00,cart,458
			2024-04-18 10:00:00,2024-04-18 10:10:00,payment,550
			2024-04-18 10:00:00,2024-04-18 10:10:00,trade,550
			2024-04-18 10:00:00,2024-04-18 10:10:00,good_detail,992
		 */

		// 4. 数据接收器-sink
		/*
		保存MySQL数据库表
-- 创建数据库
CREATE DATABASE IF NOT EXISTS gmall_report ;
USE gmall_report ;

-- 窗口表
CREATE TABLE IF NOT EXISTS tbl_page_view_report (
    window_start VARCHAR(255),
    window_end VARCHAR(255),
    page_id VARCHAR(255),
    pv long,
    PRIMARY KEY (window_start, window_end, page_id)
) ;

REPLACE INTO tbl_page_view_report(window_start, window_end, page_id, pv) VALUES (?, ?, ?, ?) ;
		 */
		SinkFunction<String> jdbcSink = JdbcSink.sink(
			"REPLACE INTO tbl_page_view_report(window_start, window_end, page_id, pv) VALUES (?, ?, ?, ?) ",
			(statement, value) -> {
				// value -> 2024-04-18 10:00:00,2024-04-18 10:10:00,mine,126
				String[] split = value.split(",");
				// 设置值
				statement.setObject(1, split[0]);
				statement.setObject(2, split[1]);
				statement.setObject(3, split[2]);
				statement.setObject(4, split[3]);
			},
			JdbcExecutionOptions.builder()
				.withBatchSize(1000)
				.withBatchIntervalMs(200)
				.withMaxRetries(5)
				.build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
				.withUrl("jdbc:mysql://node101:3306/gmall_report")
				.withDriverName("com.mysql.cj.jdbc.Driver")
				.withUsername("root")
				.withPassword("123456")
				.build()
		);
		stream36.addSink(jdbcSink);

		// 5. 触发执行-execute
		env.execute("RealtimePageViewDemo") ;
	}

}
